package org.goldfish.mapreduce;


//import lombok.extern.log4j.Log4j;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;
import org.goldfish.mapreduce.orders.OrderKey;
import org.goldfish.mapreduce.orders.OrderMapper;
import org.goldfish.mapreduce.orders.OrderReducer;
import org.goldfish.mapreduce.wordcount.WordCountMapper;
import org.goldfish.mapreduce.wordcount.WordCountReducer;

import java.io.IOException;
import java.util.Arrays;

@Slf4j
public class Main {
    public static void main(String[] args)
            throws IOException, ClassNotFoundException, InterruptedException {

        // 1. 加载Hadoop配置
        Configuration conf = new Configuration();
        log.info("Hadoop Configuration: {}", conf);

        // 2. 解析命令行参数
        log.info("Command line arguments: {}", Arrays.toString(args));
        String[] real_args;
        switch (args.length) {
            case 2:
                real_args = args;
                break;
            case 3:
                if (!args[0].equals(Main.class.toString())) {
                    log.error("Invalid command line arguments: started with {},", args[0]);
                    log.error("Which did not match class name {}", Main.class.getName());
                    System.exit(1);
                    return;
                }
                real_args = Arrays.copyOfRange(args, 1, args.length);
                break;
            default:
                log.error("Invalid command line arguments: got only {}", args.length);
                System.exit(1);
                return;
        }
        String[] otherArgs = new GenericOptionsParser(conf, real_args).getRemainingArgs();
        log.warn("i.e. {}", Arrays.toString(otherArgs));

        String inputPath = real_args[0];
        String outputPath = real_args[1];
        log.info("Input path: {}, Output path: {}", inputPath, outputPath);

        // 3. 创建Job对象
        Job job = Job.getInstance(conf, "Word Count");

        // 4. 设置主类（用于找到Job JAR包）
        job.setJarByClass(Main.class);

        specifyJobHandler(job);

        setInterfaceClass(job);

        // 8. 设置输入/输出格式（默认TextInputFormat和TextOutputFormat）
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        // 9. 设置输入路径和输出路径
        TextInputFormat.addInputPath(job, new Path(inputPath));
        TextOutputFormat.setOutputPath(job, new Path(outputPath));
        log.info("Job submitted successfully.");

        // 10. 提交任务并等待完成
        boolean success = job.waitForCompletion(true);
        if (!success) {
            System.out.println("Job failed!");
            System.exit(1);
        }
    }

    private static void setInterfaceClass(Job job) {
        // 6. 设置Mapper输出类型（可选，如果与最终输出相同可以省略）
        job.setMapOutputKeyClass(OrderKey.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 7. 设置最终输出类型（Reducer输出）
        job.setOutputKeyClass(OrderKey.class);
        job.setOutputValueClass(IntWritable.class);
    }

    private static void specifyJobHandler(Job job) {
        // 5. 设置Mapper和Reducer类
        job.setMapperClass(OrderMapper.class);
        job.setReducerClass(OrderReducer.class);
    }

}
